package com.tzg157.fitness.service.realtime.ads;

import cn.hutool.core.date.DateUtil;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.functions;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

@Service
public class ADSCourseTopNServiceImpl {

    @Resource
    private ConcurrentHashMap<String,Object> queueMap;
    public void topn(SparkSession spark) {
        String now = DateUtil.format(new Date(),"yyyy-MM-dd");
        //获取pv数据
        String pvSql = "select course_id, count(1) as pv_count from fitness.dwd_course_pv group by course_id";
        Dataset<Row> pvDataset = spark.sql(pvSql);

        //获取rate数据
        String rateSql = "select course_id, avg(rate) as avg_rate, count(1) as rate_count from fitness.dwd_course_rate group by course_id";
        Dataset<Row> rateDataset = spark.sql(rateSql);
        //获取reserve数据
        String reserveSql = "select course_id, count(1) as reserve_count from fitness.dwd_course_reserve group by course_id";
        Dataset<Row> reserveDataset = spark.sql(reserveSql);
//        spark.sql("select course_id,count(1) as pv_count from ods_course_pv group by course_id order by pv_count desc limit 10").show();

        // 定义权重
        double pvWeight = 0.4;
        double avgRateWeight = 0.4; // 平均评分的权重
        double rateCountWeight = 0.3; // 评分数量的权重
        double reserveWeight = 1;

        // 合并三个数据集
        Dataset<Row> joinedDataset = pvDataset
                .join(rateDataset, "course_id", "left_outer")
                .join(reserveDataset, "course_id", "left_outer")
                .na().fill(0); // 填充缺失值为0

        // 计算综合得分
        Dataset<Row> scoredDataset = joinedDataset.withColumn("score",
                functions.expr(String.format("(pv_count * %f + avg_rate * %f * %f + rate_count * %f * %f + reserve_count * %f) / (%f + %f * %f + %f * %f + %f)",
                        pvWeight, avgRateWeight, avgRateWeight, rateCountWeight, rateCountWeight, reserveWeight,
                        pvWeight, avgRateWeight, rateCountWeight, rateCountWeight, rateCountWeight, reserveWeight)));

        // 找出Top 10的course_id
        Dataset<Row> topN = scoredDataset.orderBy(functions.desc("score")).limit(10);

        // 显示结果
        topN.printSchema();
        topN.show();
        System.out.println(topN.collectAsList());
        List<Object> msgList = new ArrayList<>();
        topN.collectAsList().forEach(row->{
            msgList.add(row.toString());
        });

        // 将结果保存到Hive中
//        topN.write().mode("overwrite").saveAsTable("fitness.ads_course_topn");
        queueMap.put("adsCourseTopN",msgList);
    }

    public static void main(String[] args) {
        System.out.println(DateUtil.format(new Date(),"yyyy-MM-dd"));
    }

}
